package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.MySQLUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

public class TableProcessFunction  extends ProcessFunction<JSONObject,JSONObject> {
    // 因为要输出到侧输出流，需要定义一个侧输出流标签
    private OutputTag<JSONObject> outputTag;

    // 在内存中存储表配置对象
    private Map<String, TableProcess> tableProcessMap = new HashMap<>();

    //表示目前内存中已经存在的HBase表
    private Set<String> existsTables = new HashSet<>();

    //生命Phoenix连接
    private Connection connection=null;


    public TableProcessFunction(OutputTag<JSONObject> outputTag){
        this.outputTag = outputTag;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化配置表信息
        initTableProcessMap();

        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        this.connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);

        // 数据配置可能发生变化，需要开启一个定时任务，每隔一段时间从配置表中查询一次数据
        Timer timer = new Timer();
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                // 查询MySql表
                initTableProcessMap();
            }
        },100000,100000);
    }

    //每过来一个元素，方法执行一次
    @Override
    public void processElement(JSONObject jsonObject, Context context, Collector<JSONObject> collector) throws Exception {
        String table = jsonObject.getString("table");
        String type = jsonObject.getString("type");

        // 问题修复： 如果使用MaxWell的Bootstrap 同步历史数据，这个时候它的操作类型叫做bootstrap-insert
        if("bootstrap-insert".equals(type)){
            type ="insert";
            jsonObject.put("type",type);
        }


        // 从内存中获取当前key对象的配置信息
        if(tableProcessMap != null && tableProcessMap.size() >0) {
            String key = table + ":" + type;
            // 获取信息
            TableProcess tableProcess = tableProcessMap.get(key);
            if(tableProcess != null){
                jsonObject.put("sink_table",tableProcess.getSinkTable());
                if(tableProcess.getSinkColumns() != null && tableProcess.getSinkColumns().length() >0) {
                    filterColumn(jsonObject.getJSONObject("data"),tableProcess.getSinkColumns());
                }
            }else{
                System.out.println("No This Key: "+ key +"in mysql 配置表！！！");
            }

            if (tableProcess !=null && TableProcess.SINK_TYPE_HBASE.equalsIgnoreCase(tableProcess.getSinkType())){
                context.output(outputTag,jsonObject);
            }else if(tableProcess != null && TableProcess.SINK_TYPE_KAFKA.equalsIgnoreCase(tableProcess.getSinkType()))
                collector.collect(jsonObject);
        }


    }



    private void filterColumn(JSONObject data,String sinkColumn){
        String[] cols = sinkColumn.split(",");
        Set<Map.Entry<String,Object>> entries = data.entrySet();
        List<String> columnList = Arrays.asList(cols);
        entries.removeIf(entry -> !columnList.contains(entry.getKey()));
    }



    // 读取MySQL中配置信息，存入到内存Map中
    private void initTableProcessMap(){
        System.out.println("更新配置的处理信息");
        //查询MySQL中的配置表数据
        List<TableProcess> tableProcessList = MySQLUtil.queryList("select * from table_process",TableProcess.class,true);
        for(TableProcess tableProcess:tableProcessList) {
            String sourceTable = tableProcess.getSourceTable();
            String operateType = tableProcess.getOperateType();
            String sinkTable = tableProcess.getSinkTable();
            String sinkType = tableProcess.getSinkType();
            String sinkColumns = tableProcess.getSinkColumns();

            String sinkPK = tableProcess.getSinkPk();

            String sinkExt = tableProcess.getSinkExtend();
            //拼接Key
            String key = sourceTable + ":" + operateType;
            tableProcessMap.put(key, tableProcess);

            System.out.println(tableProcess);
            // 如果是向HBase中保存的表，需要判断内存中维护的set是否存在
            if ("insert".equals(operateType) && "hbase".equals(sinkType)){
                boolean notExist = existsTables.add(sourceTable);
                // 如果不存在，则需要Phoenix中新创建
                if(notExist){
                    // 检查在Phoenix中是否存在
                    // 有可能已经存在，只不过是应用缓存被清空，导致当前没有缓存，这种情况是不需要创建表的
                    // 在Phoenix中的确不存在的，需要将表创建出来
                    checkTable(sinkTable,sinkColumns,sinkPK,sinkExt);
                }
            }
        }

        if(tableProcessMap == null || tableProcessMap.size() ==0){
            throw  new RuntimeException("缺少信息处理");
        }
    }

    private void checkTable(String tableName,String fields,String pk,String ext) {
        //主键不存在，则给定默认值
        if(pk == null){
            pk = "id";
        }

        if(ext== null){
            ext = "";
        }

        //拼接建表语句  命名空间.table
        StringBuilder createSql = new StringBuilder("create table if not exists " +
                GmallConfig.HBASE_SCHEMA + "." + tableName + "( ");

        String[] fieldsArr = fields.split(",");
        for (int i=0;i< fieldsArr.length;i++){
            String field = fieldsArr[i];
            if(pk.equals(field)){
                createSql.append(field).append(" varchar primary key ");
            }else{
                createSql.append("info.").append(field).append(" varchar");
            }

            if (i < fieldsArr.length-1){
                createSql.append(",");
            }
        }

        createSql.append(")");
        createSql.append(ext);

        PreparedStatement ps = null;

        System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>" + (connection == null));

        String sql = createSql.toString();
        try{
            if (this.connection == null){
                this.connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
            }
            System.out.println("<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<" + (connection == null));
            //执行建表语句
            System.out.println(sql);
            ps = connection.prepareStatement(sql);
            ps.execute();
            ps.close();
        }catch (SQLException e){
            e.printStackTrace();
            throw new RuntimeException("建表失败！！！");
        }finally {
            if (ps != null){
                try {
                    ps.close();
                }catch (SQLException e){
                    e.printStackTrace();
                    throw new RuntimeException("Phoenix 建表失败");
                }
            }
        }



    }
}
